package com.example.day04.controller.sideoutputs;

import com.example.day04.controller.bean.WaterSemsor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 旁路输出
 * 测输出流
 */
public class Demo1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        DataStreamSource<String> source = env.socketTextStream("node3", 7777);

        SingleOutputStreamOperator<WaterSemsor> map = source.map((String s) -> {
            String[] split = s.split(",");
            return new WaterSemsor(split[0], Long.parseLong(split[1]), split[2]);
        });

//        source.map(new RichMapFunction<String, WaterSemsor>() {
//            @Override
//            public WaterSemsor map(String s) throws Exception {
//                String[] split = s.split(",");
//                return new WaterSemsor(split[0], Long.parseLong(split[1]), split[2]);
//            }
//        });

//        String[] split = line.split(",");
//        if (split.length < 3) {
//            throw new IllegalArgumentException("Line does not contain enough elements: " + line);
//        }
//        return new WaterSemsor(split[0], Long.parseLong(split[1]), split[2]);


        OutputTag s1 = new OutputTag("s1", Types.POJO(WaterSemsor.class));
        OutputTag s2 = new OutputTag("s2", Types.POJO(WaterSemsor.class));
        SingleOutputStreamOperator<WaterSemsor> process = map.process(new ProcessFunction<WaterSemsor, WaterSemsor>() {
            @Override
            public void processElement(WaterSemsor waterSemsor, ProcessFunction<WaterSemsor, WaterSemsor>.Context context, Collector<WaterSemsor> collector) throws Exception {
                String id = waterSemsor.getId();
                if (id.equals("s1")) {
                    context.output(s1, waterSemsor);
                } else if (id.equals("s2")) {
                    context.output(s2, waterSemsor);
                } else {
                    collector.collect(waterSemsor);
                }
            }
        });

        process.print("zhao主流");
//        process.printToErr()
        process.getSideOutput(s1).print("zhaos1");
        process.getSideOutput(s2).print("zhaos2");

        env.execute();


    }
}
